package com.icreate.storm.bolt;

import com.icreate.storm.methods.TupleAccumulate.TupleAccumulator;
import com.icreate.storm.methods.TupleAccumulate.indicator.IndicatorAccumulator;
import com.icreate.storm.methods.TupleAccumulate.indicator.IndicatorStaticAccumulator;
import com.icreate.storm.methods.indicators.GetId;
import com.icreate.storm.methods.indicators.Indicator;
import org.apache.commons.lang.Validate;
import org.apache.storm.mongodb.bolt.AbstractMongoBolt;
import org.apache.storm.mongodb.common.QueryFilterCreator;
import org.apache.storm.mongodb.common.mapper.MongoMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.TupleUtils;
import org.bson.Document;
import org.bson.conversions.Bson;
import scala.Serializable;
import scala.Tuple2;

import java.security.Key;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Created by feng on 2017/5/14.
 */
public class MongoBufferUpdateBolt extends AbstractMongoBolt {
    private final int bufferSize = 150;
    private GetId getId;
    private QueryFilterCreator queryCreator;
    private Map<String, Tuple2<Bson, TupleAccumulator>> buffer = new LinkedHashMap<>(bufferSize);
    private AccumulatorPool accumulatorPool;
    private Long count = 0L;

    private boolean upsert;  //The default is false.

    private static class AccumulatorPool implements Serializable {
        private IndicatorStaticAccumulator[] pool;
        private int surplus;

        public AccumulatorPool(int bufferSize, Map<String, String[]> indicatorsFields) {
            pool = new IndicatorStaticAccumulator[bufferSize];
            for (int i = 0; i < pool.length; i++) {
                pool[i] = new IndicatorStaticAccumulator(indicatorsFields);
            }
            this.surplus = bufferSize;
        }

        public IndicatorStaticAccumulator getOne() {
            if (surplus == 0)
                return null;
            surplus--;
            return pool[surplus];
        }

        public void reset() {
            for (IndicatorStaticAccumulator accumulator : this.pool) {
                accumulator.reset();
            }
            this.surplus = this.pool.length;
        }
    }

    public MongoBufferUpdateBolt(String url, String collectionName, QueryFilterCreator queryCreator, Map<String, String[]> indicatorsFields, GetId getId) {
        super(url, collectionName);
        Validate.notNull(queryCreator, "QueryFilterCreator can not be null");
        Validate.notNull(indicatorsFields, "indicatorsFields can not be null");

        this.getId = getId;
        accumulatorPool = new AccumulatorPool(bufferSize, indicatorsFields);
        this.queryCreator = queryCreator;


    }

    @Override
    public void execute(Tuple tuple) {
        if (TupleUtils.isTick(tuple)) {
            return;
        }
//        try {
        if (buffer.size() >= bufferSize || count > 10000) {
            runUpdate();
            accumulatorPool.reset();
            buffer.clear();
            count = 0L;
        }
        addOne(tuple);
        count++;
        this.collector.ack(tuple);
//        } catch (Exception e) {
//            this.collector.reportError(e);
//            this.collector.fail(tuple);
//        }

    }

    private void runUpdate() {

        for (Map.Entry<String, Tuple2<Bson, TupleAccumulator>> entry : this.buffer.entrySet()) {
            Document doc = entry.getValue()._2().toDocument();
            Bson filter = entry.getValue()._1();
            if (doc == null)
                continue;
            try {
                mongoClient.update(filter, doc, upsert);
            } catch (Exception e) {
                this.collector.reportError(e);
            }
        }

    }

    private void addOne(Tuple tuple) {
        String id = this.getId.getId(tuple);
        Tuple2<Bson, TupleAccumulator> entry;
        if (this.buffer.containsKey(id)) {
            entry = this.buffer.get(id);
            entry._2().add(tuple);
        } else {
            entry = new Tuple2<>(queryCreator.createFilter(tuple), accumulatorPool.getOne());
            entry._2().add(tuple);
            this.buffer.put(id, entry);
        }
    }

    public MongoBufferUpdateBolt withUpsert(boolean upsert) {
        this.upsert = upsert;
        return this;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}

